package com.smsc.headend.connector.config;


import cn.hutool.core.convert.Convert;
import com.smsc.headend.connector.netty.AnsiServer;
import com.smsc.headend.connector.netty.COSEMServer;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.ConnectorTask;
import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.assertj.core.util.Lists;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.Order;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.LinkedBlockingQueue;

@Configuration
@Slf4j
@Order(2)
public class CommonConfig {
    @Autowired
    AdminClient kafkaAdminClient;
    @Autowired
    KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;


    @Value("${ports.dcu_port:6603}")
    String dcuPort;

    @Value("${ports.p2p_port:6604}")
    String p2pPort;

    @Value("${ports.ansi_p2p_port:6703}")
    String ansiP2pPort;

    @Value("${client.establish.limiter.size:100}")
    String limiterSize;

    @Bean(name = "nioServer", initMethod = "start", destroyMethod = "stop")
    public COSEMServer nioServer() {
        return new COSEMServer(Convert.toInt(dcuPort));
    }

    @Bean(name = "nioGprsServer", initMethod = "start", destroyMethod = "stop")
    public COSEMServer nioGprsServer() {
        return new COSEMServer(Convert.toInt(p2pPort));
    }

    @Bean(name = "ansiGprsServer", initMethod = "start", destroyMethod = "stop")
    public AnsiServer anisGprsServer() {
        return new AnsiServer(Convert.toInt(ansiP2pPort));
    }

    @Bean("tokenPool")
    public LinkedBlockingQueue<String> getTokenPool() {
        return new LinkedBlockingQueue<String>(Integer.valueOf(limiterSize));
    }

    @Bean
    public MeterRegistry registry() {
        return new SimpleMeterRegistry();
    }

    @Bean("dlmsTimer")
    public Timer dlmsTimer(MeterRegistry registry) {
        return Timer.builder("dlms.packet.accept").description("accept dlms packet").register(registry);
    }


    @Bean("taskAllInboundCounter")
    public Counter taskAllInboundCounter(MeterRegistry registry) {
        return Counter.builder("con.task.allin.counter").description("task all inbound to CON").register(registry);
    }

    @Bean("taskInboundCounter")
    public Counter taskInboundCounter(MeterRegistry registry) {
        return Counter.builder("con.task.inbound.counter").description("task inbound to CON").register(registry);
    }

    @Bean("taskOutboundCounter")
    public Counter taskOutboundCounter(MeterRegistry registry) {
        return Counter.builder("con.task.outbound.counter").description("task outbound return to UTE").register(registry);
    }

    @Bean("deviceTicketPool")
    public LinkedBlockingQueue<Integer> deviceTicketPool(@Value("${con.task.device.concurrency:100}") Integer conNumber) {
//        Integer conNumber = Convert.toInt(concurrentDeviceNumber);
        LinkedBlockingQueue<Integer> queue =  new LinkedBlockingQueue<>(conNumber);
        for (int i = 0; i < conNumber; i++) {
            queue.add(i);
        }
        return queue;
    }

    @Bean("deviceTaskQueue")
    public LinkedBlockingQueue<ConnectorTask> deviceTaskQueue(@Value("${con.task.device.queueSize:100000}") Integer queueSize) {
        LinkedBlockingQueue<ConnectorTask> queue =  new LinkedBlockingQueue<>(queueSize);
        return queue;
    }

    @Bean
    public ApplicationRunner runnerCreateTopic() {
        return args ->
        {
            DescribeTopicsResult result = kafkaAdminClient.describeTopics(Lists.newArrayList(
                    TaskKafkaTopic.TASK_SEND_TO_CON
//                    , TaskKafkaTopic.FTP_COLLECTION_TO_CON
            ));

            List<NewTopic> newTopicList = new ArrayList();
            result.values().forEach((s, topicDescriptionKafkaFuture) -> {
                try {
                    topicDescriptionKafkaFuture.get();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } catch (ExecutionException e) {
                    log.info("{} not exists ,create by CON", s);
                    NewTopic topic = new NewTopic(s, 3, (short) 1);
                    newTopicList.add(topic);
                }
            });
            kafkaAdminClient.createTopics(newTopicList);
            //wait for kafka setting config
            if (newTopicList.size() > 0) {
                Thread.sleep(10000);
            }


            kafkaListenerEndpointRegistry.getAllListenerContainers().forEach(messageListenerContainer -> {
                if (!messageListenerContainer.isRunning()) {
                    log.info("start {}", messageListenerContainer);
                    messageListenerContainer.start();
                } else {
                    log.info("resume {}", messageListenerContainer);
                    messageListenerContainer.resume();
                }
            });
        };
    }
}
